package com.ebs.md.fxmonitor.marketmonitor;

/**
 * Perforce Revision:
 *  $File: //depot/MarketData/fxmonitor/main/src/main/java/com/ebs/md/fxmonitor/marketmonitor/CrossMarketListener.java $
 *  $Revision: #24 $  $Change: 116237 $  $DateTime: 2012/10/25 14:20:22 $  $Author: Arik.Sher $
 */

import java.math.BigDecimal;
import java.util.Calendar;
import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import org.apache.commons.collections.map.LRUMap;
import org.springframework.util.Assert;

import com.ebs.commons.comp.IParsedPayload;
import com.ebs.market.monitor.MarketMonitor.CrossMarketEvent;
import com.ebs.market.monitor.MarketMonitor.EventType;
import com.ebs.market.monitor.MarketMonitor.MarketIdentfier;
import com.ebs.market.monitor.MarketMonitor.QuoteSide;
import com.ebs.marketdata.dictionary.MatchingType;
import com.ebs.md.api.IDetailedBookEntry;
import com.ebs.md.api.OrderSide;
import com.ebs.md.api.Rate;


public class CrossMarketListener
extends AbstractMonitorListenerAndPublisher
{
	protected static final int tasksPoolSize = 32; // each pool handles a MarketID(pair+MatchingType)

	
	protected ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(tasksPoolSize);
	// map between marketId string to runnable future result
	protected ConcurrentMap<String, ScheduledFuture<?>> crossMarketFuturesMap;

	// solving PTR 14470 - when CrossMarket, do not notify again on same quote.
	// note that the current impl, use the same LRU pool for all pairs(marketIds)
	protected int    maxRecentQuoteIds = 512;
	// holding quote_ids (String), mapped to long(milliseconds) - the time that alert was sent
	protected Map<String, Long> mruCrossedMarketQuoteIds;
	

	public CrossMarketListener() {
		super();
		crossMarketFuturesMap = new ConcurrentHashMap<String, ScheduledFuture<?>>(32);

		scheduledExecutorService = Executors.newScheduledThreadPool(tasksPoolSize);
	}

	public void setScheduledExecutorService(ScheduledExecutorService scheduledExecutorService) {
		this.scheduledExecutorService = scheduledExecutorService;
	}
	
	public void setMaxRecentQuoteIds(int maxRecentQuoteIds) {
		this.maxRecentQuoteIds = maxRecentQuoteIds;
	}

	public void init() {
		super.init();
		Assert.notNull(scheduledExecutorService, "'scheduledExecutorService' cannot be null");
		
		mruCrossedMarketQuoteIds = 
			// using synchronized map, over the LRUMap
			// to solve PTR 14881, where NullPointerException might be thrown.
			// see http://mail-archives.apache.org/mod_mbox/commons-user/200606.mbox/%3C20060618082325.255590@gmx.net%3E
			Collections.synchronizedMap(
				(Map<String, Long>)
				new LRUMap(maxRecentQuoteIds)
			)
		;
	}
	
	public void destroyMe() {
		Set<String> tmpSet = new HashSet<String>(crossMarketFuturesMap.keySet());

		for ( String currMarketId : tmpSet ) {
			stopExistingTask(currMarketId);
		}
	}

	/**
	 * 
	 * assuming that payload is 'TOBC' (TopOfBook Change one).
	 */
	protected void handleParsedPayload(IParsedPayload payload) {
		boolean thresholdExceeds = thresholdExceeds(payload);
		
		String marketId = CommonPayloadParse.getMarketIdAsStr(payload);

		// a new TOB change arrived, make the previous task stop(if exists)
		stopExistingTask(marketId);
		
		boolean isNewEvent = false;
		if ( thresholdExceeds ) {
			isNewEvent = isNewEvent(payload);
		}

		if ( thresholdExceeds && isNewEvent ) {
//		if ( true ) { // use only in Debug Mode
			// schedule a task for that marketId
			long thesholdMillis = getThresholdInterval(payload);
			ScheduledFuture<?> scheduledFuture = scheduledExecutorService.schedule
				( new CrossMarketTimedPublisher(this, payload, mruCrossedMarketQuoteIds)
				, thesholdMillis, TimeUnit.MILLISECONDS)
				;
			crossMarketFuturesMap.put(marketId, scheduledFuture);
		}
	}

	/**
	 * Solving PTR 14470 - do no notify more than once on same quote.
	 * in future: maybe in future per floor.
	 * 
	 * @param payload
	 * @return
	 */
	protected boolean isNewEvent(IParsedPayload payload) {
		List<IDetailedBookEntry> bestBidEntries   = (List<IDetailedBookEntry>)payload.getParsedField("book_entries_of_top_bid");
		List<IDetailedBookEntry> bestOfferEntries = (List<IDetailedBookEntry>)payload.getParsedField("book_entries_of_top_offer") ;
		
		for ( IDetailedBookEntry curr : bestBidEntries ) {
			if ( mruCrossedMarketQuoteIds.containsKey(curr.getQuoteId()) ) {
				// found quote id on MRU. it is not new event
				return false;
			}
		}
		for ( IDetailedBookEntry curr : bestOfferEntries ) {
			if ( mruCrossedMarketQuoteIds.containsKey(curr.getQuoteId()) ) {
				// found quote id on MRU. it is not new event
				return false;
			}
		}
		
		// default true
		return true;
	}

	protected void stopExistingTask(String marketId) {
		ScheduledFuture<?> future = crossMarketFuturesMap.get(marketId);
		
		if ( future != null ) {
			boolean canceled = future.cancel(true);
			if ( !canceled ) {
				logger.debug(String.format("did not manage to cancel task for market id: '%s'", marketId));
			}
			crossMarketFuturesMap.remove(marketId);
		}
	}
	
	@Override
	protected boolean thresholdExceeds(IParsedPayload payload) {
		Rate   bidTopRate = CommonPayloadParse.  getBidTopRate(payload);
		Rate offerTopRate = CommonPayloadParse.getOfferTopRate(payload);
		
		if ( bidTopRate == null || offerTopRate == null ) {
			return false;
		}
		
		BigDecimal ratesDiff = CommonTools.calcRatesDiff(bidTopRate, offerTopRate);
		
		if ( !CommonTools.isGreaterEqual(ratesDiff, BigDecimal.ZERO) ) {
			return false;
		}
		
		final MatchingType matchingType = CommonPayloadParse.getMatchingType(payload);
		// solve PTR 14738
		if (    false
				|| MatchingType.Fix.equals(matchingType)
				|| MatchingType.CM .equals(matchingType) 
		) {
			return false;
		}

		// solving PTR 14628, negative or zero amount not should be raised.
		long   relevantAmount = 0;
		if ( CommonPayloadParse.isActionAdd(payload) ) {
			relevantAmount = CommonPayloadParse.getOrderAmount(payload);
		} else {
			// action is REMOVE
			OrderSide orderSide = CommonPayloadParse.getOrderSide(payload);
			List<IDetailedBookEntry> bestEntries = CommonPayloadParse.getBestEntriesListBySide(payload, orderSide);
			IDetailedBookEntry be = CommonTools.safeExtractedFirst(bestEntries);

			if ( be != null ) {
				relevantAmount = be.getAmount();
			} else {
				logger.warn("not likely to enter here...");
				
			}
		}
		
		if ( relevantAmount <= 0 ) {
			return false;
		}
		
		return true;
	}

	protected EventMessageCreator getCreator(IParsedPayload payload) {
		return new CrossMarketJmsMessageCreator(payload, getThresholdInterval(payload));
	}

	protected long getThresholdInterval(IParsedPayload payload) {
		String  baseCurrencyIso = CommonPayloadParse. getBaseCurrencyIso(payload);
		String localCurrencyIso = CommonPayloadParse.getLocalCurrencyIso(payload);
		return getThresholdInterval(baseCurrencyIso, localCurrencyIso);
	}

	protected long getThresholdInterval(String baseCurrencyIso, String localCurrencyIso) {
		if ( eventConfigProvider != null ) {
			return eventConfigProvider.getCrossMarketIntervalThreshold(baseCurrencyIso, localCurrencyIso);
		}
		return EVENT_CROSS_MARKET_DEFAULT_INTERVAL_THRESHOLD;
	}
	
	static class CrossMarketTimedPublisher
	implements Runnable
	{
		protected AbstractMonitorListenerAndPublisher publisher;
		protected IParsedPayload                      payload;
		protected Map<String, Long>                   mruQuoteIds;

		public CrossMarketTimedPublisher(AbstractMonitorListenerAndPublisher publisher, IParsedPayload payload, Map<String, Long> mruQuoteIds) {
			super();
			this.publisher = publisher;
			this.payload = payload;
			this.mruQuoteIds = mruQuoteIds;
		}

		public void run() {
			try {
				Thread.sleep(10); // calling the sleep just to catch the InterruptedException(which could be raised within the stopExisting Task)
				
				{
					// storing the quote id at MRU
					// TODO: small performance : call createEventObject somehow only once, and also publish it.
					final CrossMarketJmsMessageCreator cmjmc = new CrossMarketJmsMessageCreator(payload, 4321);
					final CrossMarketEvent eventObj = cmjmc.createEventObject(payload);
					mruQuoteIds.put(eventObj.getQuoteId(), System.currentTimeMillis() );
				}
				
				publisher.doPublish(payload);
				
			} catch (InterruptedException e) {
				publisher.logger.debug("this task been interrupted. cancelling publication of CrossMarket event.");
			} catch (Throwable e) {
				publisher.logger.warn("unexpected exception... could not publish CrossMarket event", e);
			}
		}
		
	}
	
	static class CrossMarketJmsMessageCreator 
	extends AbstractEventMessageCreator<CrossMarketEvent>
	{
		protected long thresholdInterval;
		
		public CrossMarketJmsMessageCreator(IParsedPayload payload, long thresholdInterval) {
			super(payload);
			this.thresholdInterval = thresholdInterval;
		}

		public EventType getEventType() {
			return EventType.CROSS_MARKET;
		}

		/**
		 * @see com.ebs.md.fxmonitor.orders.TopOfBooksChangeListener#createBasicNotificationPayload
		 * @see com.ebs.md.fxmonitor.orders.TopOfBooksChangeListener#createNotificationPayloadByOrder
		 */
		public CrossMarketEvent createEventObject(IParsedPayload payload) {
			CrossMarketEvent.Builder builder = CrossMarketEvent.newBuilder();
		
			MarketIdentfier.Builder miBuilder = MarketIdentfier.newBuilder();
			miBuilder. setBaseCurrencyIso(payload.getParsedStringField( "base_ccy_iso"));
			miBuilder.setLocalCurrencyIso(payload.getParsedStringField("local_ccy_iso"));
			miBuilder.setMatchingType(    payload.getParsedIntField(   "matching_type"));
			
			builder.setMarketId(miBuilder.build());
			builder.setEventTime(eventTimeMillis);
			
			builder.  setBidRate(payload.getParsedStringField(  "bid_top_rate_str"));
			builder.setOfferRate(payload.getParsedStringField("offer_top_rate_str"));
			
			handleOrderToOutput(builder);
			
			builder.setMillisThreshold(thresholdInterval);
			
			CrossMarketEvent built = builder.build();
			return built;
		}
		
		protected void handleOrderToOutput(CrossMarketEvent.Builder builder) {
			final OrderSide orderSide = CommonPayloadParse.getOrderSide(payload);
			final QuoteSide quoteSide = CommonTools.getQuoteSide(orderSide);
			builder.setQuoteSide(quoteSide);

			// if the action is  ADD than it is quicker to extract data from payload directly.
			final boolean isActionAdd = CommonPayloadParse.isActionAdd(payload);
			final boolean useAffectingQuote = isActionAdd;
			
			if ( !useAffectingQuote ) {
				// use BookEntries on same side. choose the first entry
				List<IDetailedBookEntry> bestEntries = CommonPayloadParse.getBestEntriesListBySide(payload, orderSide);
				IDetailedBookEntry be = CommonTools.safeExtractedFirst(bestEntries);
				
				if ( be == null ) {
					// not likely scenario to enter this section...
					logger.warn("cannot populate details about the problematic quote");
				} else {
					builder.setQuoteId(be.getQuoteId());
					builder.setQuoteRate(be.getRate().toString());
					builder.setQuoteAmount(be.getAmount());
					
					Calendar quoteCal = be.getQuoteTime();
					builder.setQuoteTime(quoteCal.getTimeInMillis());
					
					final String dealCode = be.getDealCode();
					if ( dealCode != null ) {
						builder.setDealCode(dealCode);
					}
					String traderId = be.getTraderId();
					if ( traderId != null ) {
						builder.setTraderId(traderId);
					}
				}
			} else {
				// use affecting quote, which caused the TOB change.
				// it is not good case to use it when the action is REMOVE, because:
				//  1) this quote, because it was removed from OB, it caused TOB change, but it is not the one which caused the CrossMarket...
				//  2) it won't contain the amount, rate... (minor issue)
				builder.setQuoteId(    payload.getParsedStringField("order_id"));
				builder.setQuoteRate(  payload.getParsedStringField("order_rate_str"));
				builder.setQuoteAmount(payload.getParsedLongField(  "order_amount"));

				Date actionDate =  payload.getParsedDateField("action_time");
				builder.setQuoteTime(actionDate.getTime());

				String dealCode = payload.getParsedStringField("deal_code", null);
				if ( dealCode != null ) {
					builder.setDealCode(dealCode);
				}
				String traderId = payload.getParsedStringField("trader_id", null);
				if ( traderId != null ) {
					builder.setTraderId(traderId);
				}
			}
			
			String eventId = genEventId(builder.getQuoteId(), builder.getQuoteTime());
			builder.setEventId(eventId);
		}
		
	}

}
